package com.hsogoo.collate.batch;

import com.google.common.base.Splitter;
import com.google.common.collect.Lists;
import com.hsogoo.collate.domain.CollateFieldsDO;
import com.hsogoo.collate.domain.CollateResult;
import com.hsogoo.collate.domain.DynamicQuery;
import com.hsogoo.collate.enums.DiffType;
import com.hsogoo.collate.exception.CollateException;
import com.hsogoo.collate.mapper.CollateFieldsMapper;
import com.hsogoo.collate.redis.RedisClient;
import com.hsogoo.collate.service.AsyncDynamicQueryService;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.JobParameters;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.repeat.RepeatStatus;

import java.lang.reflect.Field;
import java.util.List;
import java.util.Map;
import java.util.Set;

/**
 * @Author huangsaigang
 * Created @2019-10-28.
 */
@Slf4j
@Setter
public class DoCollateTasklet implements Tasklet {

    private String TARGET_SET_KEY = "{%s}:targetSet:%s";
    private String ORIGIN_SET_KEY = "{%s}:originSet:%s";

    private CollateFieldsMapper collateFieldsMapper;
    private AsyncDynamicQueryService asyncDynamicQueryService;
    private RedisClient redisClient;

    @Override
    public RepeatStatus execute(StepContribution stepContribution, ChunkContext chunkContext) throws Exception {
        log.info("DoCollateTasklet.executing...");
        JobParameters jobParameters = stepContribution.getStepExecution().getJobExecution().getJobParameters();

        //TODO:sharding分片比对，一次性全部数据比对可能OOM以及Redis的阻塞

        //构建动态数据源查询源数据
        String dataSourceName = jobParameters.getString("dataSourceName");
        String originSql = jobParameters.getString("originSql");
        DynamicQuery dynamicQuery = new DynamicQuery();
        dynamicQuery.setDataSourceName(dataSourceName);
        dynamicQuery.setSql(originSql);
        List<Map<String, Object>> originResult = asyncDynamicQueryService.dynamicQuery(dynamicQuery);
        log.info("=============动态数据源查询源数据完成=============");

        //获取任务名和对账时间
        String taskName = jobParameters.getString("taskName");
        String collateDate = jobParameters.getString("targetCollateDate");

        //把源数据根据对账配置put到redis的集合中
        String originKeyword = jobParameters.getString("originKeyword");
        String[] originCollateValues = createOriginCollateValues(originResult, originKeyword);
        String originSetKey = String.format(ORIGIN_SET_KEY, taskName, collateDate);
        redisClient.sadd(originSetKey, originCollateValues);
        log.info("=============源数据放置到redis完成=============");

        //查询出对账文件中的目标数据
        List<CollateFieldsDO> targetResult = collateFieldsMapper.getTargetCollateFields(taskName, collateDate);

        ////把目标数据根据对账配置put到redis的集合中
        String targetKeyword = jobParameters.getString("targetKeyword");
        String targetFieldMapper = jobParameters.getString("targetFieldMapper");
        String[] targetCollateValues = createTargetCollateValues(targetResult, targetFieldMapper, targetKeyword);
        String targetSetKey = String.format(TARGET_SET_KEY, taskName, collateDate);
        redisClient.sadd(targetSetKey, targetCollateValues);
        log.info("=============目标数据放置到redis完成=============");

        Set<String> targetDiff = redisClient.sdiff(targetSetKey, originSetKey);
        log.info("目标差异数据:{}", targetDiff.size());
        List<CollateResult> targetResultList = Lists.newArrayList();
        for(String targetDiffKeyword : targetDiff){
            CollateResult targetCollateResult = new CollateResult();
            targetCollateResult.setDiffType(DiffType.TARGET.name());
            targetCollateResult.setTargetKeyword(targetDiffKeyword);
            targetResultList.add(targetCollateResult);
            log.info("目标差异数据:{}条目标存在源不存在", targetDiff.size());
        }
        Set<String> originDiff = redisClient.sdiff(originSetKey, targetSetKey);
        for(String targetDiffKeyword : targetDiff){
            CollateResult targetCollateResult = new CollateResult();
            targetCollateResult.setDiffType(DiffType.TARGET.name());
            targetCollateResult.setTargetKeyword(targetDiffKeyword);
            targetResultList.add(targetCollateResult);
            log.info("源差异数据:{}条源存在目标不存在", originDiff.size());
        }

        //TODO：都存在，但值不匹配

        log.info("=============对账完成=============");
        //TODO：记录对账差异结果
        return RepeatStatus.FINISHED;
    }

    /**
     * 根据配置的keyword做双边对账
     * keyword即两边关联的字段
     * @param originResult
     * @param originKeyword
     * @return
     */
    private String[] createOriginCollateValues(List<Map<String, Object>> originResult, String originKeyword ){
        String[] originCollateValues = new String[originResult.size()];
        int index = 0;
        for(Map<String, Object> map : originResult){
//            originCollateValues[index] = Joiner.on(",").join(Arrays.asList(map.values()));
            Object originKeywordValue = map.get(originKeyword);
            if (originKeywordValue == null) {
                log.error("源关联字段[{}]在数据库查询中不存在", originKeyword);
                throw new CollateException("源关联字段在数据库查询中不存在");
            }
            originCollateValues[index++] = (String) originKeywordValue;
        }
        return originCollateValues;
    }


    /**
     * 根据配置的keyword做双边对账
     * @param targetResult
     * @param targetFieldMapper     字段映射
     * @param targetKeyword         关联字段
     * @return
     */
    private String[] createTargetCollateValues(List<CollateFieldsDO> targetResult,String targetFieldMapper, String targetKeyword ){
        String[] targetCollateValues = new String[targetResult.size()];
        List<String> targetFields = Splitter.on(",").trimResults().splitToList(targetFieldMapper);
        if (!targetFields.contains(targetKeyword)) {
            log.error("目标关联字段[{}]在目标字段映射[{}]中不存在", targetKeyword, targetFieldMapper);
            throw new CollateException("目标关联字段在字段映射中不存在");
        }
        int suffix = targetFields.indexOf(targetKeyword) + 1;
        int index = 0;
        for(CollateFieldsDO collateFields : targetResult){
            String fieldName = "field" + suffix;
            try {
                Field field = collateFields.getClass().getDeclaredField(fieldName);
                field.setAccessible(true);
                targetCollateValues[index++] = (String) field.get(collateFields);
            } catch (Exception e) {
                log.error("目标关联字段[{}]在数据库查询中不存在", targetKeyword);
                throw new CollateException("目标关联字段在数据库查询中不存在");
            }
        }
        return targetCollateValues;
    }
}
